package com.liyghting.kafka;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author LiYiGuang
 * @date 2018/9/26 9:48
 */
@EnableKafka
@RestController
@SpringBootApplication
public class App {
	public static void main(String[] args) {
		SpringApplication.run(App.class, args);
	}

	@Autowired
	private KafkaTemplate kafkaTemplate;

	@GetMapping("send")
	public void sendKafkaMsg() {
		kafkaTemplate.setDefaultTopic("someTopic");
		kafkaTemplate.send("someTopic", "hello world");
		ProducerRecord producerRecord = new ProducerRecord("someTopic", "hello world2");
		kafkaTemplate.send(producerRecord);
		Message<String> message = new GenericMessage("hello world3");
		kafkaTemplate.send(message);
		ListenableFuture<SendResult> sendResult = kafkaTemplate.sendDefault(0, "1", "hello world4");
		sendResult.addCallback(result -> System.out.println("result = " + result.getProducerRecord().value()),
				ex -> System.out.println("ex = " + ex));

	}

	@KafkaListener(topics = "someTopic")
	public void processMessage(String content) {
		System.out.println("content = " + content);
	}
}
